package quartz.job.batch;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.beans.factory.annotation.Autowired;
import quartz.config.JobConsumingTimeListener;
import quartz.config.StepConsumingTimeListener;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public abstract class Base<I, O> {
    @Autowired
    protected JobBuilderFactory jobBuilderFactory; // 用于构建 Job
    @Autowired
    protected StepBuilderFactory stepBuilderFactory; // 用于构建 Step

    protected ItemReader<I> reader(Collection<I> items) {
        BlockingQueue<I> queue = new LinkedBlockingQueue<>(items);
        return queue::poll; // 读数据，可以用 JPA、JDBC、JMS 等方式
    }

    protected abstract O processor(I item);

    protected abstract void writer(List<? extends O> items);

    protected Step step(ItemReader<I> reader) { // 一个 Step 通常由 ItemReader(读)->ItemProcessor(处理)-ItemWriter(写) 组成
        return stepBuilderFactory.get(getClass().getName())
            .<I, O>chunk(1000) // processor 1000 条后 writer 一次
            .reader(reader).processor((ItemProcessor<I, O>) this::processor).writer(this::writer)
            .listener(new StepConsumingTimeListener())
            .build();
    }

    protected Job job(Step step) { // 一个 Job 通常由一个或多个 Step 组成
        return jobBuilderFactory.get(getClass().getName())
            .start(step)
            // .next(xxxStep()).next(xxxStep())
            .listener(new JobConsumingTimeListener())
            .build();
    }
}